package com.new_api;

import com.sun.org.apache.xpath.internal.WhitespaceStrippingElementMatcher;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class HBaseAPIDemo2 {

    private Configuration conf;
    private Connection conn;
    private Admin admin;

    @Before
    public void beforeTest() throws IOException {
        conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "node02,node03,node04");
        conn = ConnectionFactory.createConnection(conf);
        admin = conn.getAdmin();
    }

    @Test
    public void hbaseSplit() throws IOException {
        //自定义算法，产生一系列Hash散列值存储在二维数组中
        byte[][] splitKeys = {{'a', 'b'}, {'d', 'e'}};

        TableName tableName = TableName.valueOf("api_test:student2");
        TableDescriptorBuilder tableDesc = TableDescriptorBuilder.newBuilder(tableName);

        ColumnFamilyDescriptorBuilder cfDesc = ColumnFamilyDescriptorBuilder.newBuilder("info".getBytes());

        tableDesc.setColumnFamily(cfDesc.build());

        admin.createTable(tableDesc.build(), splitKeys);
    }


    /**
     * 扫描数据
     *
     * @throws Exception
     */
    @Test
    public void scanRows() throws Exception {
        String nameSapce = "api_test";
        String tableTitle = "student";
        // 获取table
        Table table = conn.getTable(TableName.valueOf(nameSapce, tableTitle));

        Scan scan = new Scan();
        // 控制扫描的行限制
        scan.withStartRow(Bytes.toBytes("1500100001"));
        scan.withStopRow(Bytes.toBytes("1500100003"));

        ResultScanner scanner = table.getScanner(scan);
        Result result = null;
        while ((result = scanner.next()) != null) {
            String id = Bytes.toString(result.getRow());
            System.out.println(id);
            String name = Bytes.toString(result.getValue("info".getBytes(), "name".getBytes()));
            String age = Bytes.toString(result.getValue("info".getBytes(), "age".getBytes()));
            String gender = Bytes.toString(result.getValue("info".getBytes(), "gender".getBytes()));
            String clazz = Bytes.toString(result.getValue("info".getBytes(), "clazz".getBytes()));
            System.out.println("name:" + name + ",age:" + age + ",gender:" + gender + ",clazz:" + clazz);
        }

        table.close();
    }


    /**
     * get读取一行数据
     */
    @Test
    public void getTest() throws Exception {
        // 要先获取表
        TableName tableName = TableName.valueOf("api_test:student");
        Table table = conn.getTable(tableName);
        Get get = new Get("1500100999".getBytes());
        // 可选，读取某一列的数据
        // get.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"));

        // 获取查询结果
        Result result = table.get(get);
        String id = Bytes.toString(result.getRow());
        System.out.println(id);
        String name = Bytes.toString(result.getValue("info".getBytes(), "name".getBytes()));
        String age = Bytes.toString(result.getValue("info".getBytes(), "age".getBytes()));
        String gender = Bytes.toString(result.getValue("info".getBytes(), "gender".getBytes()));
        String clazz = Bytes.toString(result.getValue("info".getBytes(), "clazz".getBytes()));
        System.out.println("name:" + name + ",age:" + age + ",gender:" + gender + ",clazz:" + clazz);

        // 从结果中获取Cell
        Cell[] cells = result.rawCells();
        for (Cell cell : cells) {
            // 从cell中获取字节数据，并封装到字符串中
            String value = Bytes.toString(CellUtil.cloneValue(cell));
            System.out.print(value + ",");
        }

        // 关闭资源
        table.close();

    }

    /**
     * 删除数据
     */
    @Test
    public void deleteTest() throws Exception {
        // 要先获取表
        TableName tableName = TableName.valueOf("api_test:student");
        Table table = conn.getTable(tableName);

        Delete delete = new Delete("1500101000".getBytes());

        delete.addColumn("info".getBytes(), "gender".getBytes());

        table.delete(delete);
        table.close();
    }

    /**
     * 批量put一批数据到HBase
     */
    @Test
    public void putAllTest() throws Exception {
        // 要先获取表
        TableName tableName = TableName.valueOf("api_test:student");
        Table table = conn.getTable(tableName);

        BufferedReader br = new BufferedReader(new FileReader("data/students.txt"));

        String line;
        List<Put> list = new ArrayList<>();
        while ((line = br.readLine()) != null) {
            String[] split = line.split(",");
            // rowkey,info:name,info:age, info:gender, info:clazz
            Put put = new Put(split[0].getBytes());
            put.addColumn("info".getBytes(), "name".getBytes(), split[1].getBytes());
            put.addColumn("info".getBytes(), "age".getBytes(), split[2].getBytes());
            put.addColumn("info".getBytes(), "gender".getBytes(), split[3].getBytes());
            put.addColumn("info".getBytes(), "clazz".getBytes(), split[4].getBytes());
            list.add(put);
        }

        table.put(list);

        table.close();
    }

    /**
     * put一条数据到hbase表，添加、修改
     */
    @Test
    public void putTest() throws Exception {
        TableName tableName = TableName.valueOf("api_test", "student");
        // ....
        Table table = conn.getTable(tableName);

        Put put = new Put("1500100005".getBytes());

        put.addColumn("info".getBytes(), "name".getBytes(), Bytes.toBytes("xm"));

        table.put(put);

        table.close();
    }


    @After
    public void afterTest() throws IOException {
        admin.close();
        conn.close();
    }


}
